package com.develop.collection.middle.sync;

import com.develop.collection.middle.sensors.SensorsData;
import com.develop.collection.util.Entry;
import com.develop.jdbc.jdbc.DataRow;
import com.develop.jdbc.jdbc.session.Session;
import com.develop.jdbc.jdbc.session.SessionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author fenny
 * @since 2019/09/05
 */
public class EventsSyncLocal {

    private static final Logger log = LoggerFactory.getLogger(SensorsData.class);

    private static final String TABLE_NAME = "kilimall_events";

    private static final AtomicInteger SUCCESS_COUNT = new AtomicInteger();

    private static final AtomicInteger FILTER_COUNT = new AtomicInteger();

    private static final AtomicInteger CONVERT_COUNT = new AtomicInteger();

    private static void execute() {
        log.info("begin-采集");
        int currTotal = 376058244;
        int pageSize = 100000;
        int totalPage = (currTotal + pageSize - 1) / pageSize;
        ExecutorService pagePool = Executors.newFixedThreadPool(10);
        ExecutorService pool = Executors.newFixedThreadPool(1000);
        int beginPage = 1;
        try {
            CountDownLatch pageLatch = new CountDownLatch(totalPage);
            for (int i = beginPage; i <= totalPage; i++) {
                int finalI = i;
                pagePool.submit(() -> {
                    int offset = (finalI - 1) * pageSize;
                    //读取数据
                    Session source = SessionFactory.getSession("local-mysql");
                    List<DataRow> list = source.query("select event,user_id,sku_id,num from kilimall_events", offset, pageSize);
                    source.close();
                    CountDownLatch latch = new CountDownLatch(list.size());
                    list.forEach(row -> {
                        pool.submit(() -> {
                            Session target = null;
                            try {
                                target = SessionFactory.getSession("centos-local-mysql");
                                target.insert(TABLE_NAME, row);
                                SUCCESS_COUNT.incrementAndGet();
                            } catch (Exception e) {
                                log.error("", e);
                            } finally {
                                latch.countDown();
                                if (target != null) {
                                    target.close();
                                }

                            }
                        });
                    });
                    try {
                        latch.await();
                    } catch (InterruptedException e) {
                        log.error("page=" + finalI, e);
                    } finally {
                        pageLatch.countDown();
                    }
                    log.info("造完一个阶段={}", finalI);
                });
            }
            pageLatch.await();
        } catch (Exception e) {
            log.error("", e);
        } finally {
            pool.shutdown();
            pagePool.shutdown();
            log.info("原数据量={},过滤量={},写入量={},转换量={}", currTotal, FILTER_COUNT.get(), SUCCESS_COUNT.get(), CONVERT_COUNT.get());
            log.info("end-采集");
        }
    }

    public static void main(String[] args) {
        Entry.entry(EventsSyncLocal::execute);
    }

}
